package com.reluxer.filter;

import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Activate(group = CommonConstants.CONSUMER)
public class TPMonitorFilter implements Filter {

    private class TPMonitor implements Runnable {
        private final long EXPIRE_MILLIS = 60 * 1000;

        private final String method;

        // key: 方法调用发生时间 value: 方法调用耗时
        private final Map<Long, Long> timeRecords = new ConcurrentHashMap<>();

        TPMonitor(String method) {
            this.method = method;
            // 每隔5s执行一次run
            Executors.newSingleThreadScheduledExecutor()
                    .scheduleWithFixedDelay(this, 0, 5, TimeUnit.SECONDS);
        }

        public void record(long timeStamp, long timeUsed) {
            timeRecords.put(timeStamp, timeUsed);
        }

        private void removeExpiredEntry() {
            if (timeRecords.size() == 0) {
                return;
            }
            long earliestTimeStamp = System.currentTimeMillis() - EXPIRE_MILLIS;
            Set<Long> timestamps = timeRecords.keySet();
            for (Long timestamp : timestamps) {
                if (timestamp < earliestTimeStamp) {
                    timeRecords.remove(timestamp);
                }
            }
        }

        @Override
        public void run() {
            removeExpiredEntry();
            if (timeRecords.size() == 0) {
                System.out.println("method " + method + " has no records yet...");
                return;
            }
            Collection<Long> timeUsedColl = timeRecords.values();
            Object[] timeUsedSortedArray = timeUsedColl.stream().sorted().toArray();
            int total = timeUsedColl.size();
            int tp90Index = (int) (total * 0.9);
            int tp99Index = (int) (total * 0.99);

            System.out.println("method " + method + " TP90 is " + timeUsedSortedArray[tp90Index] + "ms, in " + tp90Index + "/" + total);
            System.out.println("method " + method + " TP99 is " + timeUsedSortedArray[tp99Index] + "ms, in " + tp99Index + "/" + total);
        }
    }

    private final Map<String, TPMonitor> methodMonitor = new ConcurrentHashMap<>();

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        String methodName = invocation.getMethodName();
        long start = System.currentTimeMillis();
        Result result = invoker.invoke(invocation);
        int timeUsed = (int) (System.currentTimeMillis() - start);
        TPMonitor tpMonitor = methodMonitor.get(methodName);
        if (tpMonitor == null) {
            synchronized (methodMonitor) {
                tpMonitor = methodMonitor.get(methodName);
                if (tpMonitor == null) {
                    tpMonitor = new TPMonitor(methodName);
                    methodMonitor.put(methodName, tpMonitor);
                }
            }
        }
        tpMonitor.record(start, timeUsed);
        return result;
    }
}
